package cn.cqsztech.mq.active.util;

import cn.hutool.core.util.ObjectUtil;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTextMessage;

import javax.jms.*;
import java.util.concurrent.TimeUnit;

/**
 * ccmars
 * 2022/3/2
 **/
public class ActiveMQUtil {
    public static void main(String[] args) throws JMSException {
    }

    private static final String URL = "tcp://192.168.248.226:61616";
    private static final String USER = "admin";
    private static final String PWD = "admin";

    public static void send() {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USER, PWD, URL);
        try {
            QueueConnection connection = activeMQConnectionFactory.createQueueConnection();
            QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            QueueSender sender = session.createSender(() -> "topic");
            ActiveMQTextMessage activeMQTextMessage = new ActiveMQTextMessage();
            activeMQTextMessage.setText("hi");
            sender.send(activeMQTextMessage);
            session.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static TopicConnection getConnection() {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USER, PWD, URL);
        try {
            TopicConnection topicConnection = activeMQConnectionFactory.createTopicConnection();
            return topicConnection;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("连接创建失败！" + e.getMessage());
        }
    }

    public static void publish(Topic topic, Message message) {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USER, PWD, URL);
        try {
            TopicConnection topicConnection = activeMQConnectionFactory.createTopicConnection();
            TopicSession session = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            MessageProducer producer = session.createProducer(topic);
            producer.send(message);
            session.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static Message subscribe(Topic topic) throws JMSException {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USER, PWD, URL);
        TopicSession session = null;
        try {
            TopicConnection topicConnection = activeMQConnectionFactory.createTopicConnection();
            session = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            MessageConsumer consumer = session.createConsumer(topic);
            topicConnection.start();
            Message receive = consumer.receive();
            return receive;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (session != null)
                session.close();
        }
        return null;
    }

    public static void closeSessionGracefully(Session session) {
        try {
            if (session != null) {
                session.close();
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public static void receive() {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USER, PWD, URL);
        try {
            QueueConnection connection = activeMQConnectionFactory.createQueueConnection();
            connection.start();
            QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            QueueReceiver receiver = session.createReceiver(() -> "topic");
            while (true) {
                ActiveMQTextMessage receive = (ActiveMQTextMessage) receiver.receive(1000);
                if (ObjectUtil.isNotEmpty(receive))
                    System.out.println(receive.getText());
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
